package org.etlt.extract;

import org.etlt.EtltRuntimeException;
import org.etlt.job.JobContext;
import org.etlt.job.PkRange;

import javax.sql.DataSource;
import java.sql.*;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

public class DatabaseExtractor extends Extractor {
    private final DatabaseExtractSetting setting;

    private DataSource dataSource;

    private Connection connection;

    private PreparedStatement statement;

    private ResultSet resultSet;

    private boolean ready = false;

    public DatabaseExtractor(DatabaseExtractSetting setting) {
        this.setting = setting;
        this.setName(setting.getName());
    }

    public DatabaseExtractSetting getSetting() {
        return setting;
    }

    public DataSource getDataSource() {
        return dataSource;
    }

    public void setDataSource(DataSource dataSource) {
        this.dataSource = dataSource;
    }

    public Connection getConnection() {
        return connection;
    }

    public void setConnection(Connection connection) {
        this.connection = connection;
    }

    public PreparedStatement getStatement() {
        return statement;
    }

    public void setStatement(PreparedStatement statement) {
        this.statement = statement;
    }

    public ResultSet getResultSet() {
        return resultSet;
    }

    public void setResultSet(ResultSet resultSet) {
        this.resultSet = resultSet;
    }

    public boolean isReady() {
        return ready;
    }

    public void setReady(boolean ready) {
        this.ready = ready;
    }

    @Override
    public synchronized void init(JobContext context) {
        if (!ready) {
            try {
                if (this.setting.getDatasource() != null) {
                    this.dataSource = context.getResource(this.setting.getDatasource());
                    this.connection = this.dataSource.getConnection();
                    String dql = populate(getSetting().getDql(), context);
                    this.statement = this.connection.prepareStatement(dql);
                    resultSet = this.statement.executeQuery();
                    ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
                    int columnCount = resultSetMetaData.getColumnCount();
                    if (this.setting.isAutoResolve() && getColumns().size() == 0) {
                        for (int i = 0; i < columnCount; i++) {
                            getColumns().add(resultSetMetaData.getColumnLabel(i + 1));
                        }
                    }
                }
                ready = true;
            } catch (SQLException e) {
                throw new EtltRuntimeException("Extractor init error: " + getName(), e);
            }
        }
    }


    protected String populate(String dql, JobContext context) {
        String startFlag = "${";
        String endFlag = "}";
        StringBuilder stringBuilder = new StringBuilder();
        int start = 0;
        int indexOfStart = dql.indexOf(startFlag, start);
        int indexOfEnd = dql.indexOf(endFlag, start);
        for (; indexOfStart != -1 && indexOfEnd != -1;
             indexOfStart = dql.indexOf(startFlag, start), indexOfEnd = dql.indexOf(endFlag, start)) {
            stringBuilder.append(dql, start, indexOfStart);
            start = indexOfEnd + 1;
            String propertyName = dql.substring(indexOfStart + startFlag.length(), indexOfEnd);
            stringBuilder.append(context.getPropertyValue(startFlag + propertyName + endFlag));
        }
        stringBuilder.append(dql.substring(start));
        return stringBuilder.toString();
    }
    @Override
    public boolean isClosed() {
        return false;
    }


    @Override
    public void extract(JobContext context) {
        if(!ready)
            init(context);
        try {
            if (this.connection == null || this.connection.isClosed()) {
                this.connection = this.dataSource.getConnection();
                this.statement = this.connection.prepareStatement(this.setting.getDql());
                resultSet = this.statement.executeQuery();
            }
            synchronized (this) {
                if (this.resultSet.next()) {
                    if (this.skip < this.setting.getSkip()) {
                        this.skip++;
                        this.index++;
                        extract(context);
                    } else {
                        Map<String, Object> rowData = new HashMap<>();
                        ResultSetMetaData resultSetMetaData = resultSet.getMetaData();
                        for (int i = 0; i < getColumns().size(); i++) {
                            if (getColumns().contains(resultSetMetaData.getColumnLabel(i + 1)))
                                rowData.put(getColumns().get(i),
                                        DatabaseUtil.getObject(resultSet, i + 1,
                                                resultSetMetaData.getColumnType(i + 1))
                                );
                        }
                        Entity entity = new Entity(index++, rowData);
                        context.setEntity(this.setting.getName(), entity);
                    }
                } else {
                    context.removeEntity(this.setting.getName());
                }
            }
        } catch (SQLException e) {
            throw new EtltRuntimeException(e);
        }
    }


    @Override
    public List<String> getColumns() {
        return this.setting.getColumns();
    }

    @Override
    public DatabaseExtractor createInstance() {
        DatabaseExtractor extractor = new DatabaseExtractor(this.setting);
        extractor.dataSource = this.dataSource;
        return extractor;
    }

    @Override
    public void close() {
        close(resultSet, statement, connection);
    }

    @Override
    public PkRange getPkFullRange() {
        throw new UnsupportedOperationException("not supported by DatabaseExtractor.");
    }

    @Override
    public String toString() {
        StringBuilder detail = new StringBuilder(super.toString());
        detail.append(",dql: ");
        detail.append(getSetting().getDql());
        return detail.toString();
    }
}
